package com.learn.log.kafka.listener;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaLogConsumer {

	private static final Logger posLog = LoggerFactory.getLogger("pos");
	private static final Logger erpLog = LoggerFactory.getLogger("erp");
	private static final Logger memberLog = LoggerFactory.getLogger("member");

	@KafkaListener(topics = "log-member")
	public void listenMemberLog(ConsumerRecord<?, ?> record) {
		logHandler(record, memberLog);
	}

	@KafkaListener(topics = "log-erp")
	public void listenErpLog(ConsumerRecord<?, ?> record) {
		logHandler(record, erpLog);
	}

	@KafkaListener(topics = "log-pos")
	public void listenPosLog(ConsumerRecord<?, ?> record) {
		logHandler(record, posLog);
	}

	private void logHandler(ConsumerRecord<?, ?> record, Logger log) {
		String logLevel = record.key().toString();
		String logMessage = record.value().toString();
		log.info(logMessage);
		if (!StringUtils.equals(logLevel, Level.INFO.toString())) {
			log.error(logMessage);
		}
	}
}
